<?php


namespace EchoPHP\Rabbitmq\Query;


use EchoPHP\Facades\RabbitMq;

class QueueSubscriber {
    protected $rabbitmqConnection;

    public function __construct(string $name) {
        $this->rabbitmqConnection = RabbitMq::connect($name);
    }

    /**
     * 消费者消费
     * @param callable $dealMessage 消息处理逻辑
     *                                  该函数签名为 function(\AMQPEnvelope $message):bool
     * @param string $queueName 要消费的队列名称
     * @param bool $requeueWhenNack 消费确认失败是否重新入列
     * @param string $consumeTag 消费者标识
     * @throws \AMQPChannelException
     * @throws \AMQPConnectionException
     * @throws \AMQPEnvelopeException
     * @throws \AMQPExchangeException
     * @throws \AMQPQueueException
     */
    public function subscribe(callable $dealMessage, string $queueName = '', bool $requeueWhenNack = false, string $consumeTag = '') {
        if (!is_callable($dealMessage)) {
            throw new \Exception('订阅回调未定义');
        }
        $queue = $this->rabbitmqConnection->getQueue($queueName);

        $flag = AMQP_NOPARAM;

        $callback_func = function (\AMQPEnvelope $message, \AMQPQueue $q) use ($dealMessage, $requeueWhenNack) {
            $dealResult = $dealMessage($message);
            if ($dealResult) {
                $q->ack($message->getDeliveryTag());
            } else {
                $q->nack($message->getDeliveryTag(), $requeueWhenNack ? AMQP_REQUEUE : AMQP_NOPARAM);
            }
        };

        $queue->consume($callback_func, $flag, $consumeTag);
    }
}